﻿#ifdef HAVE_CONFIG_H
#include <config.h>
#endif


#include "BinaryUpwardProtocol.h"
#include <hadoop/SerialUtils.hh>
#include "MsgDefines.h"

namespace HadoopPipes {
	BinaryUpwardProtocol::BinaryUpwardProtocol(QIODevice* _stream) {
		stream = new HadoopUtils::FileOutStream();
		HADOOP_ASSERT(stream->open(_stream), "problem opening stream");
	}
	
	void BinaryUpwardProtocol::authenticate(const std::string &responseDigest) {
		serializeInt(AUTHENTICATION_RESP, *stream);
		serializeString(responseDigest, *stream);
		stream->flush();
	}

	void BinaryUpwardProtocol::output(const std::string& key, const std::string& value) {
		serializeInt(OUTPUT, *stream);
		serializeString(key, *stream);
		serializeString(value, *stream);
	}

	void BinaryUpwardProtocol::partitionedOutput(int reduce, const std::string& key,
		const std::string& value) {
			serializeInt(PARTITIONED_OUTPUT, *stream);
			serializeInt(reduce, *stream);
			serializeString(key, *stream);
			serializeString(value, *stream);
	}

	void BinaryUpwardProtocol::status(const std::string& message) {
		serializeInt(STATUS, *stream);
		serializeString(message, *stream);
	}



	void BinaryUpwardProtocol::progress(float progress) {
		serializeInt(PROGRESS, *stream);
		serializeFloat(progress, *stream);
		stream->flush();
	}

	void BinaryUpwardProtocol::done() {
		serializeInt(DONE, *stream);
	}

	void BinaryUpwardProtocol::registerCounter(int id, const std::string& group, 
		const std::string& name) {
			serializeInt(REGISTER_COUNTER, *stream);
			serializeInt(id, *stream);
			serializeString(group, *stream);
			serializeString(name, *stream);
	}

	void BinaryUpwardProtocol::incrementCounter(const TaskContext::Counter* counter, 
		uint64_t amount) {
			serializeInt(INCREMENT_COUNTER, *stream);
			serializeInt(counter->getId(), *stream);
			serializeLong(amount, *stream);
	}

	BinaryUpwardProtocol::~BinaryUpwardProtocol() throw () {
		delete stream;
	}

}